package beautiful.butterfly.drds.data_exchange.transport.record_queue;


import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.transport.record.TerminateRecord;
import beautiful.butterfly.drds.data_exchange.transport.record.column.Record;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;

import java.util.Collection;


/**
 * 统计和限速都在这里
 */
@Slf4j
public abstract class RecordQueue
{


    private static Boolean isFirstPrint = true;
    protected int taskGroupId;
    protected int capacity;
    protected int byteCapacity;
    protected long byteSpeed; // bps: bytes/s
    protected long recordSpeed; // tps: records/s
    protected long flowControlInterval;
    protected volatile boolean isClosed = false;
    protected Configuration configuration = null;
    protected volatile long waitReaderTime = 0;
    protected volatile long waitWriterTime = 0;
    private Communication communication;

    private Communication lastCommunication = new Communication();

    public RecordQueue(final Configuration configuration)
    {
        //channel的queue里默认record为1万条。原来为512条
        int capacity = configuration.getInt(
                Constants.transport_channel_capacity, 2048);
        long byteSpeed = configuration.getLong(
                Constants.transport_channel_speed_byte, 1024 * 1024);
        long recordSpeed = configuration.getLong(
                Constants.transport_channel_speed_record, 10000);

        if (capacity <= 0)
        {
            throw new IllegalArgumentException(String.format(
                    "通道容量[%d]必须大于0.", capacity));
        }

        synchronized (isFirstPrint)
        {
            if (isFirstPrint)
            {
                RecordQueue.log.info("RecordQueue set byte_speed_limit to " + byteSpeed
                        + (byteSpeed <= 0 ? ", No bps activated." : "."));
                RecordQueue.log.info("RecordQueue set record_speed_limit to " + recordSpeed
                        + (recordSpeed <= 0 ? ", No tps activated." : "."));
                isFirstPrint = false;
            }
        }

        this.taskGroupId = configuration.getInt(
                Constants.container_taskgroup_id);
        this.capacity = capacity;
        this.byteSpeed = byteSpeed;
        this.recordSpeed = recordSpeed;
        this.flowControlInterval = configuration.getLong(
                Constants.transport_channel_flowcontrolinterval, 1000);
        //channel的queue默认大小为8M，原来为64M
        this.byteCapacity = configuration.getInt(
                Constants.transport_channel_capacity_byte, 8 * 1024 * 1024);
        this.configuration = configuration;
    }

    public void close()
    {
        this.isClosed = true;
    }

    public void open()
    {
        this.isClosed = false;
    }

    public boolean isClosed()
    {
        return isClosed;
    }

    public int getTaskGroupId()
    {
        return this.taskGroupId;
    }

    public int getCapacity()
    {
        return capacity;
    }

    public long getByteSpeed()
    {
        return byteSpeed;
    }

    public Configuration getConfiguration()
    {
        return this.configuration;
    }

    public void setCommunication(final Communication communication)
    {
        this.communication = communication;
        this.lastCommunication.reset();
    }

    public void push(final Record record)
    {
        Validate.notNull(record, "record不能为空.");
        this.put(record);
        this.statPush(1L, record.getByteSize());
    }

    public void pushTerminateRecord(final TerminateRecord terminateRecord)
    {
        Validate.notNull(terminateRecord, "record不能为空.");
        this.put(terminateRecord);


    }

    public void pushAll(final Collection<Record> recordCollection)
    {
        Validate.notNull(recordCollection);
        Validate.noNullElements(recordCollection);
        //
        this.addAll(recordCollection);
        this.statPush(recordCollection.size(), this.getRecordCollectionByteSize(recordCollection));
    }

    public Record pull()
    {
        Record record = this.take();
        this.statPull(1L, record.getByteSize());
        return record;
    }

    public void pullAll(final Collection<Record> recordCollection)
    {
        Validate.notNull(recordCollection);
        this.drainTo(recordCollection);
        this.statPull(recordCollection.size(), this.getRecordCollectionByteSize(recordCollection));
    }

    protected abstract void put(Record record);

    protected abstract void addAll(Collection<Record> recordCollection);

    protected abstract Record take();

    protected abstract void drainTo(Collection<Record> recordCollection);

    public abstract int size();

    public abstract boolean isEmpty();

    public abstract void clear();

    private long getRecordCollectionByteSize(final Collection<Record> recordCollection)
    {
        long recordCollectionByteSize = 0;
        for (final Record record : recordCollection)
        {
            recordCollectionByteSize += record.getByteSize();
        }
        return recordCollectionByteSize;
    }

    private void statPush(long recordSize, long byteSize)
    {
        communication.increaseCounter(CommunicationTool.read_succeed_records,
                recordSize);
        communication.increaseCounter(CommunicationTool.read_succeed_bytes,
                byteSize);
        //在读的时候进行统计waitCounter即可，因为写（pull）的时候可能正在阻塞，但读的时候已经能读到这个阻塞的counter数

        communication.setLongCounter(CommunicationTool.wait_reader_time, waitReaderTime);
        communication.setLongCounter(CommunicationTool.wait_writer_time, waitWriterTime);

        boolean isChannelByteSpeedLimit = (this.byteSpeed > 0);
        boolean isChannelRecordSpeedLimit = (this.recordSpeed > 0);
        if (!isChannelByteSpeedLimit && !isChannelRecordSpeedLimit)
        {
            return;
        }

        long lastTimestamp = lastCommunication.getTimestamp();
        long nowTimestamp = System.currentTimeMillis();
        long interval = nowTimestamp - lastTimestamp;
        if (interval - this.flowControlInterval >= 0)
        {
            long byteLimitSleepTime = 0;
            long recordLimitSleepTime = 0;
            if (isChannelByteSpeedLimit)
            {
                long currentByteSpeed = (CommunicationTool.getTotalReadBytes(communication) -
                        CommunicationTool.getTotalReadBytes(lastCommunication)) * 1000 / interval;
                if (currentByteSpeed > this.byteSpeed)
                {
                    // 计算根据byteLimit得到的休眠时间
                    byteLimitSleepTime = currentByteSpeed * interval / this.byteSpeed
                            - interval;
                }
            }

            if (isChannelRecordSpeedLimit)
            {
                long currentRecordSpeed = (CommunicationTool.getTotalReadRecords(communication) -
                        CommunicationTool.getTotalReadRecords(lastCommunication)) * 1000 / interval;
                if (currentRecordSpeed > this.recordSpeed)
                {
                    // 计算根据recordLimit得到的休眠时间
                    recordLimitSleepTime = currentRecordSpeed * interval / this.recordSpeed
                            - interval;
                }
            }

            // 休眠时间取较大值
            long sleepTime = byteLimitSleepTime < recordLimitSleepTime ?
                    recordLimitSleepTime : byteLimitSleepTime;
            if (sleepTime > 0)
            {
                try
                {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e)
                {
                    Thread.currentThread().interrupt();
                }
            }

            lastCommunication.setLongCounter(CommunicationTool.read_succeed_bytes,
                    communication.getLongCounter(CommunicationTool.read_succeed_bytes));
            lastCommunication.setLongCounter(CommunicationTool.read_failed_bytes,
                    communication.getLongCounter(CommunicationTool.read_failed_bytes));
            lastCommunication.setLongCounter(CommunicationTool.read_succeed_records,
                    communication.getLongCounter(CommunicationTool.read_succeed_records));
            lastCommunication.setLongCounter(CommunicationTool.read_failed_records,
                    communication.getLongCounter(CommunicationTool.read_failed_records));
            lastCommunication.setTimestamp(nowTimestamp);
        }
    }

    private void statPull(long recordSize, long byteSize)
    {
        communication.increaseCounter(CommunicationTool.write_received_records, recordSize);
        communication.increaseCounter(CommunicationTool.write_received_bytes, byteSize);
    }

}
